package net.wicp.tams.commons.mq.ons;

import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;

import net.wicp.tams.commons.Conf;
import net.wicp.tams.commons.apiext.StringUtil;
import net.wicp.tams.commons.constant.Encoding;

public class SendOnsMsg {
	private final static Logger logger = LoggerFactory.getLogger(SendOnsMsg.class);
	private static volatile SendOnsMsg INSTANCE;
	private static Object lockobj = new Object();
	private static Object lockobj2 = new Object();
	private Map<String, Producer> producermap = new HashMap<String, Producer>();

	private SendOnsMsg() {
	}

	static {
		Conf.addCallBack("ons", new Conf.Callback() {
			@Override
			public void doReshConf(Properties newProperties) {
				// destroy();
				INSTANCE = null;

			}
		}, "ons.server.%s");
	}

	public static final SendOnsMsg getInstance() {
		if (INSTANCE == null) {
			synchronized (lockobj) {
				if (INSTANCE == null) {
					SendOnsMsg tempobj = new SendOnsMsg();
					logger.info("MQ发送端启动成功");
					INSTANCE = tempobj;
				}
			}
		}
		return INSTANCE;
	}

	private Producer getProducer(String producerId) {
		if (!producermap.containsKey(producerId)) {
			synchronized (lockobj2) {
				if (!producermap.containsKey(producerId)) {
					Properties properties = new Properties();
					properties.put(PropertyKeyConst.ProducerId, producerId);
					properties.put(PropertyKeyConst.AccessKey, Conf.get("ons.server.accesskey"));
					properties.put(PropertyKeyConst.SecretKey, Conf.get("ons.server.secretkey"));
					properties.put(PropertyKeyConst.ONSAddr, Conf.get("ons.server.onsaddr"));
					Producer producer = ONSFactory.createProducer(properties);
					producer.start();
					producermap.put(producerId, producer);
					return producer;
				} else {
					return producermap.get(producerId);
				}
			}
		} else {
			return producermap.get(producerId);
		}
	}

	/***
	 * 发送消息
	 * 
	 * @param producerId
	 *            producerId
	 * @param topic
	 *            主题
	 * @param tag
	 *            标签
	 * @param msg
	 *            发送的信息
	 * @return 发送结果
	 */
	public SendResult SendMsg(String producerId, String topic, String tag, String msg) {
		try {
			return SendMsg(producerId, topic, tag, msg.getBytes(Encoding.UTF8.name()));
		} catch (UnsupportedEncodingException e) {
			logger.error("发送消息失败", e);
			return null;
		}
	}

	public SendResult SendMsg(String producerId, String topic, String tag, byte[] body) {
		if (StringUtil.isNull(producerId) || StringUtil.isNull(topic) || ArrayUtils.isEmpty(body)) {
			throw new IllegalArgumentException("需要producerId、topic、msg信息");
		}
		Producer prd = getProducer(producerId);
		if (prd == null) {
			throw new IllegalArgumentException(String.format("[%s]非法，请检查producerId", producerId));
		}

		try {
			Message sendMsg = new Message(topic, StringUtil.hasNull(tag, Conf.get("ons.client.defaulttag")), body);
			SendResult sendResult = prd.send(sendMsg);
			return sendResult;
		} catch (Exception e) {
			if (e.getMessage().contains("Topic does not exist")) {// 不存在topic
				// producermap.remove(producerId);//有它，第二次会报The producer
				// group[bbbb] has been created before
			}
			logger.error(String.format("发送消息失败,消息：[{}]", new String(body)), e);
			throw new IllegalAccessError("发送消息失败,原因：" + e.getMessage());
		}
	}

	/***
	 * 给默认tag发送消息（配置文件） 发送消息
	 * 
	 * @param producerId
	 * 
	 * @param topic
	 * 
	 * @param msg
	 *            发送的信息
	 * @return 发送结果
	 */
	public SendResult SendMsg(String producerId, String topic, String msg) {
		return SendMsg(producerId, topic, null, msg);
	}

}
